package cn.iocoder.yudao.im.service.Impl;

import cn.hutool.core.collection.CollStreamUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.exception.ServiceException;
import cn.iocoder.yudao.framework.security.core.LoginUser;
import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
import cn.iocoder.yudao.im.client.IMClient;
import cn.iocoder.yudao.im.common.contant.IMConstant;
import cn.iocoder.yudao.im.common.enums.IMTerminalType;
import cn.iocoder.yudao.im.common.model.IMGroupMessage;
import cn.iocoder.yudao.im.common.model.IMUserInfo;
import cn.iocoder.yudao.im.common.util.CommaTextUtils;
import cn.iocoder.yudao.im.contant.RedisKey;
import cn.iocoder.yudao.im.domain.NoticeImGroupMember;
import cn.iocoder.yudao.im.domain.NoticeImGroupMessage;
import cn.iocoder.yudao.im.enums.MessageStatus;
import cn.iocoder.yudao.im.enums.MessageType;
import cn.iocoder.yudao.im.mapper.NoticeImGroupMessageMapper;
import cn.iocoder.yudao.im.service.NoticeImGroupMemberService;
import cn.iocoder.yudao.im.service.NoticeImGroupMessageService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.Splitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.validation.constraints.NotNull;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * <p>
 * 群消息 服务实现类
 * </p>
 *
 * @author ZCC
 * @since 2024-07-02
 */
@Service
@Slf4j
public class NoticeImGroupMessageServiceImpl extends ServiceImpl<NoticeImGroupMessageMapper, NoticeImGroupMessage> implements NoticeImGroupMessageService {


    @Resource
    private NoticeImGroupMemberService groupMemberService;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private IMClient imClient;

    @Override
    public Long sendMessage(NoticeImGroupMessage dto) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();

        // 是否在群聊里面
        NoticeImGroupMember member = groupMemberService.findByGroupAndUserId(dto.getGroupId(), loginUser.getId());
        if (Objects.isNull(member) || member.getQuit()) {
            throw new ServiceException(500, "您已不在群聊里面，无法发送消息");
        }
        // 群聊成员列表
        List<Long> userIds = groupMemberService.findUserIdsByGroupId(dto.getGroupId());
        // 不用发给自己
        userIds = userIds.stream().filter(id -> !loginUser.getId().equals(id)).collect(Collectors.toList());
        // 保存消息

        dto.setSendId(loginUser.getId());
        dto.setSendTime(LocalDateTime.now());
        dto.setSendNickName(member.getAliasName());
        if (StringUtils.isNotEmpty(dto.getAtUserIds())) {
            dto.setAtUserIds(StrUtil.join(",", dto.getAtUserIds()));
        }
        this.save(dto);
        // 群发
        IMGroupMessage<NoticeImGroupMessage> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(loginUser.getId(), 0));
        sendMessage.setRecvIds(userIds);
        sendMessage.setSendResult(false);
        sendMessage.setData(dto);
        imClient.sendGroupMessage(sendMessage);
        log.info("发送群聊消息，发送id:{},群聊id:{},内容:{}", loginUser.getId(), dto.getGroupId(), dto.getContent());
        return dto.getId();
    }

    @Override
    public void recallMessage(Long id) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        NoticeImGroupMessage msg = this.getById(id);
        if (Objects.isNull(msg)) {
            throw new ServiceException(500, "消息不存在");
        }
        if (!msg.getSendId().equals(loginUser.getId())) {
            throw new ServiceException(500, "这条消息不是由您发送,无法撤回");
        }
        long epochMilli = msg.getSendTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
        if (System.currentTimeMillis() - epochMilli > IMConstant.ALLOW_RECALL_SECOND * 1000) {
            throw new ServiceException(500, "消息已发送超过5分钟，无法撤回");
        }
        // 判断是否在群里
        NoticeImGroupMember member = groupMemberService.findByGroupAndUserId(msg.getGroupId(), loginUser.getId());
        if (Objects.isNull(member) || Boolean.TRUE.equals(member.getQuit())) {
            throw new ServiceException(500, "您已不在群聊里面，无法撤回消息");
        }
        // 修改数据库
        msg.setStatus(MessageStatus.RECALL.code());
        this.updateById(msg);
        // 群发
        List<Long> userIds = groupMemberService.findUserIdsByGroupId(msg.getGroupId());
        // 不用发给自己
        userIds = userIds.stream().filter(uid -> !loginUser.getId().equals(uid)).collect(Collectors.toList());
        msg.setType(MessageType.RECALL.code());
        String content = String.format("'%s'撤回了一条消息", member.getAliasName());
        msg.setContent(content);
        msg.setSendTime(LocalDateTime.now());

        IMGroupMessage<NoticeImGroupMessage> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(loginUser.getId(), 0));
        sendMessage.setRecvIds(userIds);
        sendMessage.setData(msg);
        sendMessage.setSendResult(false);
        sendMessage.setSendToSelf(false);
        imClient.sendGroupMessage(sendMessage);

        // 推给自己其他终端
        msg.setContent("你撤回了一条消息");
        sendMessage.setSendToSelf(true);
        sendMessage.setRecvIds(Collections.emptyList());
        sendMessage.setRecvTerminals(Collections.emptyList());
        imClient.sendGroupMessage(sendMessage);
        log.info("撤回群聊消息，发送id:{},群聊id:{},内容:{}", loginUser.getId(), msg.getGroupId(), msg.getContent());
    }


    @Override
    public void pullOfflineMessage(Long minId) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        if (!imClient.isOnline(loginUser.getId())) {
            throw new ServiceException(500, "网络连接失败，无法拉取离线消息");
        }
        // 查询用户加入的群组
        List<NoticeImGroupMember> members = groupMemberService.findByUserId(loginUser.getId());
        Map<Long, NoticeImGroupMember> groupMemberMap = CollStreamUtil.toIdentityMap(members, NoticeImGroupMember::getGroupId);
        Set<Long> groupIds = groupMemberMap.keySet();
        if (CollectionUtil.isEmpty(groupIds)) {
            // 关闭加载中标志
            this.sendLoadingMessage(false);
            return;
        }
        // 开启加载中标志
        this.sendLoadingMessage(true);
        // 只能拉取最近1个月的,最多拉取1000条
        Date minDate = DateUtils.addMonths(new Date(), -1);
        LambdaQueryWrapper<NoticeImGroupMessage> wrapper = Wrappers.lambdaQuery();
        wrapper.gt(NoticeImGroupMessage::getId, minId).gt(NoticeImGroupMessage::getSendTime, minDate).in(NoticeImGroupMessage::getGroupId, groupIds).ne(NoticeImGroupMessage::getStatus, MessageStatus.RECALL.code()).orderByDesc(NoticeImGroupMessage::getId).last("limit 1000");
        List<NoticeImGroupMessage> messages = this.list(wrapper);
        // 通过群聊对消息进行分组
        Map<Long, List<NoticeImGroupMessage>> messageGroupMap = messages.stream().collect(Collectors.groupingBy(NoticeImGroupMessage::getGroupId));
        // 退群前的消息
        List<NoticeImGroupMember> quitMembers = groupMemberService.findQuitInMonth(loginUser.getId());
        for (NoticeImGroupMember quitMember : quitMembers) {
            wrapper = Wrappers.lambdaQuery();
            wrapper.gt(NoticeImGroupMessage::getId, minId).between(NoticeImGroupMessage::getSendTime, minDate, quitMember.getQuitTime()).eq(NoticeImGroupMessage::getGroupId, quitMember.getGroupId()).ne(NoticeImGroupMessage::getStatus, MessageStatus.RECALL.code()).orderByAsc(NoticeImGroupMessage::getId).last("limit 100");
            List<NoticeImGroupMessage> noticeImGroupMessages = this.list(wrapper);
            messageGroupMap.put(quitMember.getGroupId(), noticeImGroupMessages);
            groupMemberMap.put(quitMember.getGroupId(), quitMember);
        }
        // 推送消息
        AtomicInteger sendCount = new AtomicInteger();
        messageGroupMap.forEach((groupId, FancyOfficeImGroupMessages) -> {

            // 填充消息状态
            String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
            Object o = redisTemplate.opsForHash().get(key, loginUser.getId().toString());
            long readMaxId = Objects.isNull(o) ? -1 : Long.parseLong(o.toString());
            Map<Object, Object> maxIdMap = null;
            for (NoticeImGroupMessage m : FancyOfficeImGroupMessages) {
                // 排除加群之前的消息
                NoticeImGroupMember member = groupMemberMap.get(m.getGroupId());


                if (DateUtil.compare(localDateTimeToDate(member.getCreatedTime()), localDateTimeToDate(m.getSendTime())) > 0) {
                    continue;
                }
                // 排除不需要接收的消息
                List<String> receiveIds = CommaTextUtils.asList(m.getReceiveIds());
                if (!receiveIds.isEmpty() && !receiveIds.contains(loginUser.getId().toString())) {
                    continue;
                }

                // 被@用户列表
                if (StringUtils.isNotBlank(m.getAtUserIds())) {
                    List<String> atIds = Splitter.on(",").trimResults().splitToList(m.getAtUserIds());
                    m.setAtUserList(atIds.stream().map(Long::parseLong).collect(Collectors.toList()));
                }
                // 填充状态
                m.setStatus(readMaxId >= m.getId() ? MessageStatus.READED.code() : MessageStatus.UNSEND.code());
                // 针对回执消息填充已读人数
                if (m.getReceipt()) {
                    if (Objects.isNull(maxIdMap)) {
                        maxIdMap = redisTemplate.opsForHash().entries(key);
                    }
                    int count = getReadUserIds(maxIdMap, m.getId(), m.getSendId()).size();
                    m.setReadCount(count);
                }
                //推送
                IMGroupMessage<NoticeImGroupMessage> sendMessage = getFancyOfficeImGroupMessageIMGroupMessage(loginUser, m);
                imClient.sendGroupMessage(sendMessage);
                sendCount.getAndIncrement();
            }
        });
        // 关闭加载中标志
        this.sendLoadingMessage(false);
        log.info("拉取离线群聊消息,用户id:{},数量:{}", loginUser.getId(), sendCount.get());
    }

    private static @NotNull IMGroupMessage<NoticeImGroupMessage> getFancyOfficeImGroupMessageIMGroupMessage(LoginUser loginUser, NoticeImGroupMessage vo) {
        IMGroupMessage<NoticeImGroupMessage> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(vo.getSendId(), IMTerminalType.WEB.code()));
        sendMessage.setRecvIds(Collections.singletonList(loginUser.getId()));
        sendMessage.setRecvTerminals(Collections.singletonList(0));
        sendMessage.setSendResult(false);
        sendMessage.setSendToSelf(false);
        sendMessage.setData(vo);
        return sendMessage;
    }

    @Override
    public void readMessage(Long groupId) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        // 取出最后的消息id
        LambdaQueryWrapper<NoticeImGroupMessage> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(NoticeImGroupMessage::getGroupId, groupId).orderByDesc(NoticeImGroupMessage::getId).last("limit 1").select(NoticeImGroupMessage::getId);
        NoticeImGroupMessage message = this.getOne(wrapper);
        if (Objects.isNull(message)) {
            return;
        }
        // 推送消息给自己的其他终端,同步清空会话列表中的未读数量
        NoticeImGroupMessage msgInfo = new NoticeImGroupMessage();
        msgInfo.setType(MessageType.READED.code());
        msgInfo.setSendTime(LocalDateTime.now());
        msgInfo.setSendId(loginUser.getId());
        msgInfo.setGroupId(groupId);
        IMGroupMessage<NoticeImGroupMessage> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(loginUser.getId(), 0));
        sendMessage.setSendToSelf(true);
        sendMessage.setData(msgInfo);
        sendMessage.setSendResult(true);
        imClient.sendGroupMessage(sendMessage);
        // 已读消息key
        String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
        // 原来的已读消息位置
        Object maxReadId = redisTemplate.opsForHash().get(key, loginUser.getId().toString());
        // 记录已读消息位置
        redisTemplate.opsForHash().put(key, loginUser.getId().toString(), message.getId());
        // 推送消息回执，刷新已读人数显示
        wrapper = Wrappers.lambdaQuery();
        wrapper.eq(NoticeImGroupMessage::getGroupId, groupId);
        wrapper.gt(!Objects.isNull(maxReadId), NoticeImGroupMessage::getId, maxReadId);
        wrapper.le(!Objects.isNull(maxReadId), NoticeImGroupMessage::getId, message.getId());
        wrapper.ne(NoticeImGroupMessage::getStatus, MessageStatus.RECALL.code());
        wrapper.eq(NoticeImGroupMessage::getReceipt, true);
        List<NoticeImGroupMessage> receiptMessages = this.list(wrapper);
        if (CollectionUtil.isNotEmpty(receiptMessages)) {
            List<Long> userIds = groupMemberService.findUserIdsByGroupId(groupId);
            Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
            for (NoticeImGroupMessage receiptMessage : receiptMessages) {
                int readCount = getReadUserIds(maxIdMap, receiptMessage.getId(), receiptMessage.getSendId()).size();
                // 如果所有人都已读，记录回执消息完成标记
                if (readCount >= userIds.size() - 1) {
                    receiptMessage.setReceiptOk(true);
                    this.updateById(receiptMessage);
                }
                msgInfo = new NoticeImGroupMessage();
                msgInfo.setId(receiptMessage.getId());
                msgInfo.setGroupId(groupId);
                msgInfo.setReadCount(readCount);
                msgInfo.setReceiptOk(receiptMessage.getReceiptOk());
                msgInfo.setType(MessageType.RECEIPT.code());
                sendMessage = new IMGroupMessage<>();
                sendMessage.setSender(new IMUserInfo(loginUser.getId(), 0));
                sendMessage.setRecvIds(userIds);
                sendMessage.setData(msgInfo);
                sendMessage.setSendToSelf(false);
                sendMessage.setSendResult(false);
                imClient.sendGroupMessage(sendMessage);
            }
        }
    }

    @Override
    public List<Long> findReadUsers(Long groupId, Long messageId) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        NoticeImGroupMessage message = this.getById(messageId);
        if (Objects.isNull(message)) {
            throw new ServiceException(500, "消息不存在");
        }
        // 是否在群聊里面
        NoticeImGroupMember member = groupMemberService.findByGroupAndUserId(groupId, loginUser.getId());
        if (Objects.isNull(member) || member.getQuit()) {
            throw new ServiceException(500, "您已不在群聊里面");
        }
        // 已读位置key
        String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
        // 一次获取所有用户的已读位置
        Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
        // 返回已读用户的id集合
        return getReadUserIds(maxIdMap, message.getId(), message.getSendId());
    }

    @Override
    public List<NoticeImGroupMessage> findHistoryMessage(Long groupId, Long page, Long size) {
        page = page > 0 ? page : 1;
        size = size > 0 ? size : 10;
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        Long userId = loginUser.getId();

        long stIdx = (page - 1) * size;
        // 群聊成员信息
        NoticeImGroupMember member = groupMemberService.findByGroupAndUserId(groupId, userId);
        if (Objects.isNull(member) || member.getQuit()) {
            throw new ServiceException(500, "您已不在群聊中");
        }
        // 查询聊天记录，只查询加入群聊时间之后的消息
        QueryWrapper<NoticeImGroupMessage> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(NoticeImGroupMessage::getGroupId, groupId).gt(NoticeImGroupMessage::getSendTime, member.getCreatedTime()).ne(NoticeImGroupMessage::getStatus, MessageStatus.RECALL.code()).orderByDesc(NoticeImGroupMessage::getId).last("limit " + stIdx + "," + size);

        List<NoticeImGroupMessage> messages = this.list(wrapper);
        log.info("拉取群聊记录，用户id:{},群聊id:{}，数量:{}", userId, groupId, messages.size());
        return messages;
    }

    private List<Long> getReadUserIds(Map<Object, Object> maxIdMap, Long messageId, Long sendId) {
        List<Long> userIds = new LinkedList<>();
        maxIdMap.forEach((k, v) -> {
            Long userId = Long.valueOf(k.toString());
            long maxId = Long.parseLong(v.toString());
            // 发送者不计入已读人数
            if (!sendId.equals(userId) && maxId >= messageId) {
                userIds.add(userId);
            }
        });
        return userIds;
    }

    private void sendLoadingMessage(Boolean isLoading) {
        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
        NoticeImGroupMessage msgInfo = new NoticeImGroupMessage();
        msgInfo.setType(MessageType.LOADDING.code());
        msgInfo.setContent(isLoading.toString());
        IMGroupMessage<Object> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(loginUser.getId(), 0));
        sendMessage.setRecvIds(Collections.singletonList(loginUser.getId()));
        sendMessage.setRecvTerminals(Collections.singletonList(0));
        sendMessage.setData(msgInfo);
        sendMessage.setSendToSelf(false);
        sendMessage.setSendResult(false);
        imClient.sendGroupMessage(sendMessage);
    }


    public static Date localDateTimeToDate(final LocalDateTime localDateTime) {
        if (null == localDateTime) {
            return null;
        }
        ZoneId zoneId = ZoneId.systemDefault();
        ZonedDateTime zdt = localDateTime.atZone(zoneId);
        return Date.from(zdt.toInstant());
    }


}
